package com.atguigu.app.dwd.db;

import com.atguigu.util.KafkaUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class DwdUserRegister {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnvironment = StreamTableEnvironment.create(env);

        String topicName = "topic_db";
        String groupId = "dwd_user_register";

        tableEnvironment.executeSql(KafkaUtil.getCreateTopicDB() + KafkaUtil.getKafkaDDL(topicName, groupId));


        Table userInfo = tableEnvironment.sqlQuery("select \n" +
                "`data`['id'] user_id,\n" +
                "`data`['create_time'] create_time,\n" +
                "`ts` \n" +
                "from topic_db \n" +
                "where `type`='insert' \n" +
                "and `table`='user_info'");
        tableEnvironment.createTemporaryView("user_info", userInfo);

        //tableEnvironment.sqlQuery("select * from user_info").execute().print();

        String targetTopic = "dwd_user_register";
        tableEnvironment.executeSql("create table dwd_user_register(\n" +
                "user_id STRING,\n" +
                "create_time STRING,\n" +
                "ts BIGINT\n" +
                ")" + KafkaUtil.getKafkaSinkDDL(targetTopic));
        tableEnvironment.executeSql("insert into dwd_user_register select * from user_info");
    }
}
